package com.whoami.dataplatform.tag_combine.hbase.rw;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.SequenceFile;

import java.io.IOException;
import java.util.Arrays;

/**
 * @author tzp
 * @since 2019/8/12
 */
@SuppressWarnings("Duplicates")
public class WriteTUSeqToHBase {
    private static final String TABLE_NAME = "test_tu";
    private static final String CF_DEFAULT = "cf1";

    public static void main(String[] args) throws IOException {
        Configuration config = HBaseConfiguration.create();
        config.addResource(new Path(System.getenv("HBASE_CONF_DIR") + "/conf", "hbase-site.xml"));
        writeSeqToHbase(args, config);
    }

    public static void writeSeqToHbase(String[] args, Configuration config) throws IOException {
        String input = args[0];
        int lineToRead = Integer.parseInt(args[1]);
        final int totalWrite = lineToRead;
        Configuration conf = new Configuration();

        try (SequenceFile.Reader in = new SequenceFile.Reader(conf, SequenceFile.Reader.file(new Path(input)));
             Connection connection = ConnectionFactory.createConnection(config)) {

            DataOutputBuffer readBuffer = new DataOutputBuffer();
            SequenceFile.ValueBytes valueBytes = in.createValueBytes();

            TableName tableName = TableName.valueOf(TABLE_NAME);
            Table table = connection.getTable(tableName);

            while (lineToRead > 0) {
                lineToRead--;
                readBuffer.reset();
                int eof = in.nextRaw(readBuffer, valueBytes);
                if (eof != -1) {
                    Put put = new Put(readBuffer.getData(), 0, readBuffer.getLength());//内部已array copy

                    readBuffer.reset();
                    valueBytes.writeUncompressedBytes(readBuffer);
                    byte[] bytes = Arrays.copyOf(readBuffer.getData(), readBuffer.getLength());
                    put.addColumn(CF_DEFAULT.getBytes(), null, bytes);

                    table.put(put);
                } else {
                    System.out.println("eof");
                    break;
                }
            }
        }
        System.out.println("write " + (totalWrite - lineToRead));
    }
}
